import java.util.{Date, Random, UUID}

import com.sun.xml.internal.bind.v2.TODO
import commons.conf.ConfigurationManager
import commons.constant.Constants
import commons.model.{UserInfo, UserVisitAction}
import commons.utils.{DateUtils, NumberUtils, ParamUtils, StringUtils, ValidUtils}
import net.sf.json.JSONObject
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{SaveMode, SparkSession}

import scala.collection.mutable
import scala.collection.mutable.{ArrayBuffer, ListBuffer}

object SessionStat {



  def main(args: Array[String]): Unit = {


    // onfigurationManager.config 读取的是commerce.properties里的配置信息
    // Constants.TASK_PARAMS 为限制条件对应的常量字段
    // 获取commerce.properties文件中键为Constants.TASK_PARAMS常量的值,为筛选条件
    val jsonStr: String = ConfigurationManager.config.getString(Constants.TASK_PARAMS)
    // 获取筛选条件对饮的jsonbject
    val taskParam: JSONObject = JSONObject.fromObject(jsonStr)

    //创建全局唯一的主键.uuid为每次执行都会生成一个全局的，唯一的，不可修改的128bte的标识符
    val taskUUID: String = UUID.randomUUID().toString


    // 创建sparkconf
    val sparkConf: SparkConf = new SparkConf().setAppName("session").setMaster("local[*]")

    //SparkSession 创建sparksql的环境对象(包含sparkContext)
    val sparkSession: SparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate()

    // 获取原始的动作表数据
    val actionRDD: RDD[UserVisitAction] = getOriActionRDD(sparkSession, taskParam)

    // sessionId2ActionRDD: RDD[(session_id, UserVisitAction)]，将数据转换为（session_id, UserVisitAction）的k-v结构
    val sessionId2ActionRDD: RDD[(String, UserVisitAction)] = actionRDD.map(item => (item.session_id, item))

    // session2GroupActionRDD: RDD[(session_id, Iterable[UserVisitAction])]，按照session_id 分组
    val session2GroupActionRDD: RDD[(String, Iterable[UserVisitAction])] = sessionId2ActionRDD.groupByKey()

    session2GroupActionRDD.cache()

    val sessionId2FullInfoRDD: RDD[(String, String)] = getSessionFullInfo(sparkSession, session2GroupActionRDD)

    //创建自定义累加器对象
    val sessionAccumulator: SessionAccumulator = new SessionAccumulator
    // 在sparksession中注册自定义累加器
    sparkSession.sparkContext.register(sessionAccumulator)
    // TODO : 获取需求一各个范围Session步长、访问时长占比统计
    // sessionId2FilterRDD:RDD[(sessionId,fullInfo)] 是所有符合过滤条件的数据组成的rdd
    // getSessionFilteredRDD ：实现根据限制条件对session数据进行过滤，并且完成累加器的更新
    val sessionId2FilterRDD: RDD[(String, String)] = getSessionFilteredRDD(taskParam, sessionId2FullInfoRDD, sessionAccumulator)
    sessionId2FilterRDD.collect() // 执行算子
    getSessionRatio(sparkSession, taskUUID, sessionAccumulator.value)

    // TODO : 需求二：session随机抽取
    // sessionId2FilterRDD： RDD[(sid, fullInfo)] 一个session对应一条数据，也就是一个fullInfo
    sessionRandomExtract(sparkSession, taskUUID, sessionId2FilterRDD)

    // TODO 需求三：Top10热门品类统计
    //获取符合过滤条件的 原始action数据
    val sessionId2FilterActionRDD: RDD[(String, UserVisitAction)] = sessionId2ActionRDD.join(sessionId2FilterRDD).map{
      case (sessionId,(action,fullInfo))=>
        (sessionId,action)
    }
    val top10CategoryArray: Array[(SortKey, String)] = top10PopularCategories(sparkSession, taskUUID, sessionId2FilterActionRDD)

    // TODO :需求四：Top10热门品类的Top10活跃Session统计
    top10ActionSession(sparkSession,taskUUID,sessionId2FilterActionRDD,top10CategoryArray)

  }

  // TODO :需求四：Top10热门品类的Top10活跃Session统计
  def top10ActionSession(sparkSession: SparkSession,
                         taskUUID: String,
                         sessionId2FilterActionRDD: RDD[(String, UserVisitAction)],
                         top10CategoryArray: Array[(SortKey, String)])= {
    // 1、过滤出多有点击过top品类的action

    val cidArray: Array[Long] = top10CategoryArray.map {
      case (sortKey, cidCountInfo) =>
        val cid: Long = StringUtils.getFieldFromConcatString(cidCountInfo, "\\|", Constants.FIELD_CATEGORY_ID).toLong
        cid
    }
    
    // 获得所有fullinfo里的cilckcid里包含着top的cid的列表
    val sessionId2ActionRDD: RDD[(String, UserVisitAction)] = sessionId2FilterActionRDD.filter {
      case (sessionid, filterFullInfo) =>
        cidArray.contains(filterFullInfo.click_category_id)
    }

    // 将所有的数据按session分组
    val sessionId2GroupRDD: RDD[(String, Iterable[UserVisitAction])] = sessionId2ActionRDD.groupByKey()
    val cid2SessionCountRDD: RDD[(Long, String)] = sessionId2GroupRDD.flatMap {
      case (sessionId, fullInfo) =>
        val categoryCountMap = new mutable.HashMap[Long, Long]()
        for (action <- fullInfo) {
          val cid: Long = action.click_category_id
          if (!categoryCountMap.contains(cid)) {
            categoryCountMap += (cid -> 0)
          } else {
            categoryCountMap.update(cid, categoryCountMap(cid) + 1)
          }
        }

        // 记录了一个session对于它所有点击过的品类的点击次数
        //将结果存放到数组的变量或表达式必须放在yield{}里最后位置
        for ((cid, count) <- categoryCountMap)
          yield (cid, sessionId + "=" + count)
    }

    // cid2GroupRDD: RDD[(cid, iterableSessionCount)]
    // cid2GroupRDD每一条数据都是一个categoryid和它对应的所有点击过它的session对它的点击次数
    val cid2GroupRDD: RDD[(Long, Iterable[String])] = cid2SessionCountRDD.groupByKey()
    val top10SessionRDD: RDD[Top10Session] = cid2GroupRDD.flatMap {
      case (cid, iterableSessionCount) =>
        // true: item1放在前面
        // flase: item2放在前面
        // item: sessionCount   String   "sessionId=count"
        val sortList: List[String] = iterableSessionCount.toList.sortWith((item1, item2) => {
          item1.split("=")(1).toLong > item2.split("=")(1).toLong
        }).take(10)

        sortList.map {
          case item =>
            val sessionId: String = item.split("=")(0)
            val cidClickCount: Long = item.split("=")(1).toLong

            Top10Session(taskUUID, cid, sessionId, cidClickCount)
        }
    }

    import sparkSession.implicits._

    top10SessionRDD.toDF().write
      .format("jdbc")
      .option("url", ConfigurationManager.config.getString(Constants.JDBC_URL))
      .option("user", ConfigurationManager.config.getString(Constants.JDBC_USER))
      .option("password", ConfigurationManager.config.getString(Constants.JDBC_PASSWORD))
      .option("dbtable", "top10_session_0308")
      .mode(SaveMode.Append)
      .save()
  }

  //TODO:需求三
  /**
   * 统计品类的点击次数
   * @param sessionId2FilterActionRDD
   * @return
   */
  def getClickCount(sessionId2FilterActionRDD: RDD[(String, UserVisitAction)]) = {

    //    val clickFilterRDD = sessionId2FilterActionRDD.filter{
    //      case (sessionId, action) => action.click_category_id != -1L
    //    }
    val clickFilterRDD = sessionId2FilterActionRDD.filter(item => item._2.click_category_id != -1L)

    val clickNumRDD = clickFilterRDD.map{
      case (sessionId, action) => (action.click_category_id, 1L)
    }

    clickNumRDD.reduceByKey(_+_)
  }

  /**
   *统计品类的下单次数
   * @param sessionId2FilterActionRDD
   * @return
   */
  def getOrderCount(sessionId2FilterActionRDD: RDD[(String, UserVisitAction)]) = {

    val orderFilterRDD: RDD[(String, UserVisitAction)] = sessionId2FilterActionRDD.filter(item => item._2.order_category_ids != null)
    val orderNum: RDD[(Long, Long)] = orderFilterRDD.flatMap {
      case (sessionId, action) =>
        action.order_category_ids.split(",").map(item => (item.toLong, 1L))
    }

    orderNum.reduceByKey(_+_)
  }

  /**
   * 统计品类的支付次数
   * @param sessionId2FilterActionRDD
   * @return
   */
  def getPayCount(sessionId2FilterActionRDD: RDD[(String, UserVisitAction)]) = {

    val payFilterRDD: RDD[(String, UserVisitAction)] = sessionId2FilterActionRDD.filter(item => item._2.pay_category_ids != null)
    val payNum: RDD[(Long, Long)] = payFilterRDD.flatMap{
      case(sessionId,action)=>action.pay_category_ids.split(",").map(item=>(item.toLong,1L))
    }

    payNum.reduceByKey(_+_)
  }

  /**
   * 品类数量关联
   * @param cid2CidRDD
   * @param cid2ClickCountRDD
   * @param cid2OrderCountRDD
   * @param cid2PayCountRDD
   * @return
   */
  def getFullCount(cid2CidRDD: RDD[(Long, Long)],
                   cid2ClickCountRDD: RDD[(Long, Long)],
                   cid2OrderCountRDD: RDD[(Long, Long)],
                   cid2PayCountRDD: RDD[(Long, Long)]) = {

    // leftOuterJoin做外链接,得到的值为（key,（leftvalue，option）），
    val cid2ClickInfoRDD: RDD[(Long, String)] = cid2CidRDD.leftOuterJoin(cid2ClickCountRDD).map {
      case (cid, (categoryId, option)) =>
        val cilckCount = if (option.isDefined) option.get else 0
        val aggrCount = Constants.FIELD_CATEGORY_ID + "=" + cid + "|" + Constants.FIELD_CLICK_COUNT + "=" + cilckCount

        (cid, aggrCount)
    }

    val cid2OrderInfoRDD: RDD[(Long, String)] = cid2ClickInfoRDD.leftOuterJoin(cid2OrderCountRDD).map {
      case (cid, (clickInfo, option)) =>
        val orderCount = if (option.isDefined) option.get else 0
        val aggrCount = clickInfo + "|" + Constants.FIELD_ORDER_COUNT + "=" + orderCount

        (cid, aggrCount)
    }
    val cid2PayInfoRDD: RDD[(Long, String)] = cid2OrderInfoRDD.leftOuterJoin(cid2PayCountRDD).map {
      case (cid, (orderInfo, option)) =>
        val payCount = if (option.isDefined) option.get else 0
        val aggrCouunt = orderInfo + "|" + Constants.FIELD_PAY_COUNT + "=" + payCount

        (cid, aggrCouunt)
    }

    cid2PayInfoRDD
  }

  /**
   * @param sparkSession
   * @param taskUUID
   * @param sessionId2FilterActionRDD
   * 获取符合过滤条件的 原始action数据
   * (94b36f0c8d41459fa6bfa71755359668,UserVisitAction(2021-03-23,26,94b36f0c8d41459fa6bfa71755359668,4,2021-03-23 22:45:52,null,88,67,null,null,null,null,9))
   * （date,user_id,session_id,page_id,action_time,search_keyword,click_category_id,click_product_id,order_category_ids,order_product_ids,pay_category_ids,pay_product_ids,city_id）
   * （用户点击行为的日期,用户ID,SessionID,页面ID,点击行为时间点,用户搜索关键词,点击品类ID,点击商品ID,下单品类ID,下单商品ID,支付品类ID,支付商品ID,城市ID）
   *
   */
  def top10PopularCategories(sparkSession: SparkSession,
                             taskUUID: String,
                             sessionId2FilterActionRDD: RDD[(String, UserVisitAction)])= {
    // 第一步：获取所有发生过点击、下单、付款的品类
    var cid2CidRDD = sessionId2FilterActionRDD.flatMap{
      case (sid, action)=>
        val categoryBuffer = new ArrayBuffer[(Long, Long)]()

        // 点击行为
        if(action.click_category_id != -1){
          categoryBuffer += ((action.click_category_id, action.click_category_id))
        }else if(action.order_category_ids != null){
          for(orderCid <- action.order_category_ids.split(","))
            categoryBuffer += ((orderCid.toLong, orderCid.toLong))
        }else if(action.pay_category_ids != null){
          for(payCid <- action.pay_category_ids.split(","))
            categoryBuffer += ((payCid.toLong, payCid.toLong))
        }
        categoryBuffer
    }
    //   RDD 去重
    cid2CidRDD = cid2CidRDD.distinct()

    // 第二步：统计品类的点击次数
    val cid2ClickCountRDD = getClickCount(sessionId2FilterActionRDD)
    // 第三步：统计品类的下单次数
    val cid2OrderCountRDD = getOrderCount(sessionId2FilterActionRDD)
    // 第四步：统计品类的支付次数
    val cid2PayCountRDD = getPayCount(sessionId2FilterActionRDD)

    // (4,categoryid=4|clickCount=76|orderCount=72|payCount=74)
    val cid2FullCountRDD: RDD[(Long, String)] = getFullCount(cid2CidRDD, cid2ClickCountRDD, cid2OrderCountRDD, cid2PayCountRDD)

    // 实现自定义二次排序key
    //SortKey(72,87,100),categoryid=34|clickCount=72|orderCount=87|payCount=100)
    val sortKey2FullCountRDD: RDD[(SortKey, String)] = cid2FullCountRDD.map {
      case (cid, countInfo) => {
        val clickCount = StringUtils.getFieldFromConcatString(countInfo, "\\|", Constants.FIELD_CLICK_COUNT).toLong
        val orderCount = StringUtils.getFieldFromConcatString(countInfo, "\\|", Constants.FIELD_ORDER_COUNT).toLong
        val payCount = StringUtils.getFieldFromConcatString(countInfo, "\\|", Constants.FIELD_PAY_COUNT).toLong

        val sortKey = SortKey(clickCount, orderCount, payCount)

        //SortKey(72,87,100),categoryid=34|clickCount=72|orderCount=87|payCount=100)
        (sortKey, countInfo)

      }
    }
    // false 表示倒叙
    val top10CategoryArray: Array[(SortKey, String)] = sortKey2FullCountRDD.sortByKey(false).take(10)

    val top10CategoryRDD: RDD[Top10Category] = sparkSession.sparkContext.makeRDD(top10CategoryArray).map {
      case (sortkey, countInfo) =>
        val cid: Long = StringUtils.getFieldFromConcatString(countInfo, "\\|", Constants.FIELD_CATEGORY_ID).toLong
        val clickCount = sortkey.clickCount
        val orderCount = sortkey.orderCount
        val payCount = sortkey.payCount

        Top10Category(taskUUID, cid, clickCount, orderCount, payCount)
    }

    import sparkSession.implicits._
    top10CategoryRDD.toDF().write
      .format("jdbc")
      .option("url", ConfigurationManager.config.getString(Constants.JDBC_URL))
      .option("user", ConfigurationManager.config.getString(Constants.JDBC_USER))
      .option("password", ConfigurationManager.config.getString(Constants.JDBC_PASSWORD))
      .option("dbtable", "top10_category_0308")
      .mode(SaveMode.Append) // 追加模式
      .save

    top10CategoryArray

  }

  //TODO 需求二
  /**
   *
   * @param extractPerDay    每天提取条数
   * @param dateSessionCount 天会话总数
   * @param hourCountMap     :   mutable.HashMap[小时, 会话数],小时会话数 Map(12 -> 32, 15 -> 18, 09 -> 20, 00 -> 25, 21 -> 22, 18 -> 17, 03 -> 19, 06 -> 20, 17 -> 22, 11 -> 16, 05 -> 21, 08 -> 18, 14 -> 24, 20 -> 18, 02 -> 23, 22 -> 18, 01 -> 15, 16 -> 22, 04 -> 20, 19 -> 22, 10 -> 15, 13 -> 25, 07 -> 21)
   * @param hourListMap      小时的
   * @return
   */
  def generateRandomIndexList(extractPerDay: Int,
                              dateSessionCount: Long,
                              hourCountMap: mutable.HashMap[String, Long],
                              hourListMap: mutable.HashMap[String, ListBuffer[Int]]) = {
    // （小时，该小时的session数量）
    for ((hour, hourSessionCount) <- hourCountMap) {
      // 获取一个小时要抽取多少条数据
      //一个小时要抽取的session数量 = (这个小时的session数量/这一天的session数量) * 这一天要抽取的session数量
      var hourExrCount = ((hourSessionCount / dateSessionCount.toDouble) * extractPerDay).toInt
      // 避免一个小时要抽取的数量超过这个小时的总数
      if (hourExrCount > hourSessionCount) {
        hourExrCount = hourSessionCount.toInt
      }

      val random = new Random()

      //  （date，（hour，list））
      hourListMap.get(hour) match {
        case None => hourListMap(hour) = new ListBuffer[Int]
          // 根据一个小时抽取的session数量设置随机数
          for (i <- 0 until hourExrCount) {
            var index = random.nextInt(hourSessionCount.toInt)
            // 判断小时的列表中是否有 index索引，有就重新生成
            while (hourListMap(hour).contains(index)) {
              index = random.nextInt(hourSessionCount.toInt)
            }
            hourListMap(hour).append(index)
          }
        case Some(list) =>
          for (i <- 0 until hourExrCount) {
            var index = random.nextInt(hourSessionCount.toInt)
            while (hourListMap(hour).contains(index)) {
              index = random.nextInt(hourSessionCount.toInt)
            }
            hourListMap(hour).append(index)
          }
      }
    }

  }

  /**
   *
   * @param sparkSession
   * @param taskUUID
   * @param sessionId2FilterRDD 需求一过滤后的数据
   */
  def sessionRandomExtract(sparkSession: SparkSession, taskUUID: String, sessionId2FilterRDD: RDD[(String, String)]) = {
    // 将数据转换(dateHour, filterInfo)
    val dateHour2FullInfoRDD: RDD[(String, String)] = sessionId2FilterRDD.map {
      case (sessionId, filterInfo) =>
        val startTime: String = StringUtils.getFieldFromConcatString(filterInfo, "\\|", Constants.FIELD_START_TIME)
        val dateHour: String = DateUtils.getDateHour(startTime)
        (dateHour, filterInfo)
    }

    // hourCountMap: Map[(dateHour, count)] 按照小时分组聚合值，
    // hourCountMap(2021-03-23_12,32)
    val hourCountMap: collection.Map[String, Long] = dateHour2FullInfoRDD.countByKey()

    // 创建data[hour,count]对象，每一天内每个小时的session数
    val dateHourCountMap = new mutable.HashMap[String, mutable.HashMap[String, Long]]()

    for ((dateHour, count) <- hourCountMap) {
      val date: String = dateHour.split("_")(0)
      val hour: String = dateHour.split("_")(1)

      //dateHourCountMap=Map(2021-03-23 -> Map(12 -> 32, 15 -> 18, 09 -> 20, 00 -> 25, 21 -> 22, 18 -> 17, 03 -> 19, 06 -> 20, 17 -> 22, 11 -> 16, 05 -> 21, 08 -> 18, 14 -> 24, 20 -> 18, 02 -> 23, 22 -> 18, 01 -> 15, 16 -> 22, 04 -> 20, 19 -> 22, 10 -> 15, 13 -> 25, 07 -> 21))
      dateHourCountMap.get(date) match {
        // 如果匹配没有date键，则创建dateHourCountMap(date) 的键，值为map（string,long)的空对象，
        case None => dateHourCountMap(date) = new mutable.HashMap[String, Long]()
          // dateHourCountMap(天,(小时，值))
          dateHourCountMap(date) += (hour -> count)
        // 如果存在
        case Some(map) =>
          dateHourCountMap(date) += (hour -> count)
      }
    }

    // 解决问题一：
    //         一共抽取100条数据
    //        一共有多少天： dateHourCountMap.size
    //        一天抽取多少条：100 / dateHourCountMap.size
    val extractPerDay = 100 / dateHourCountMap.size

    // 解决问题二： 一天有多少session：dateHourCountMap(date).values.sum
    // 解决问题三： 一个小时有多少session：dateHourCountMap(date)(hour)
    // dateHourExtractIndexListMap(string,(string,list))
    val dateHourExtractIndexListMap = new mutable.HashMap[String, mutable.HashMap[String, ListBuffer[Int]]]()
    //dateHourCountMap=Map(2021-03-23 -> Map(12 -> 32, 15 -> 18, 09 -> 20, 00 -> 25, 21 -> 22, 18 -> 17, 03 -> 19, 06 -> 20, 17 -> 22, 11 -> 16, 05 -> 21, 08 -> 18, 14 -> 24, 20 -> 18, 02 -> 23, 22 -> 18, 01 -> 15, 16 -> 22, 04 -> 20, 19 -> 22, 10 -> 15, 13 -> 25, 07 -> 21))
    for ((date, hourCountMap) <- dateHourCountMap) {
      // 计算一天内所有的session总数
      val dateSessionCount = hourCountMap.values.sum

      dateHourExtractIndexListMap.get(date) match {
        case None => dateHourExtractIndexListMap(date) = new mutable.HashMap[String, ListBuffer[Int]]()
          generateRandomIndexList(extractPerDay, dateSessionCount, hourCountMap, dateHourExtractIndexListMap(date))
        case Some(map) =>
          generateRandomIndexList(extractPerDay, dateSessionCount, hourCountMap, dateHourExtractIndexListMap(date))
      }

      // 到目前为止，我们获得了每个小时要抽取的session的index

      // 广播大变量，提升任务性能
      val dateHourExtractIndexListMapBd = sparkSession.sparkContext.broadcast(dateHourExtractIndexListMap)

      //提取会话RDD
      val dateHour2GroupRDD: RDD[(String, Iterable[String])] = dateHour2FullInfoRDD.groupByKey()

      val extractSessionRDD: RDD[SessionRandomExtract] = dateHour2GroupRDD.flatMap {
        case (dateHour, iterableFullInfo) =>
          val date: String = dateHour.split("_")(0)
          val hour: String = dateHour.split("_")(1)

          // 需要提取的列表索引
          val extractList: ListBuffer[Int] = dateHourExtractIndexListMapBd.value.get(date).get(hour)
          //SessionRandomExtract:Session随机抽取表
          val extractSessionArrayBuffer = new ArrayBuffer[SessionRandomExtract]()

          var index = 0
          for (fullInfo <- iterableFullInfo) {
            if (extractList.contains(index)) {
              val sessionid: String = StringUtils.getFieldFromConcatString(fullInfo, "\\|", Constants.FIELD_SESSION_ID)
              val startTime: String = StringUtils.getFieldFromConcatString(fullInfo, "\\|", Constants.FIELD_START_TIME)
              val searchKeywords: String = StringUtils.getFieldFromConcatString(fullInfo, "\\|", Constants.FIELD_SEARCH_KEYWORDS)
              val clickCategoryIds: String = StringUtils.getFieldFromConcatString(fullInfo, "\\|", Constants.FIELD_CLICK_CATEGORY_IDS)
              val extractSession: SessionRandomExtract = SessionRandomExtract(taskUUID, sessionid, startTime, searchKeywords, clickCategoryIds)
              extractSessionArrayBuffer += extractSession
            }
            index += 1
          }
          extractSessionArrayBuffer
      }

      import sparkSession.implicits._
      extractSessionRDD.toDF().write
        .format("jdbc")
        .option("url", ConfigurationManager.config.getString(Constants.JDBC_URL))
        .option("user",ConfigurationManager.config.getString(Constants.JDBC_USER))
        .option("password", ConfigurationManager.config.getString(Constants.JDBC_PASSWORD))
        .option("dbtable", "session_extract_0325")
        .mode(SaveMode.Append)
        .save()


    }
  }


  // TODO 需求一
  /**
   *
   * @param sparkSession
   * @param taskUUID
   * @param value 累加器的value值
   */
  def getSessionRatio(sparkSession: SparkSession, taskUUID: String, value: mutable.HashMap[String, Int]): Unit = {
    // session_count 作为分母，需转换为double
    val session_count = value.getOrElse(Constants.SESSION_COUNT, 1).toDouble

    // 获取对应的访问时长的值
    val visit_length_1s_3s = value.getOrElse(Constants.TIME_PERIOD_1s_3s, 0)
    val visit_length_4s_6s = value.getOrElse(Constants.TIME_PERIOD_4s_6s, 0)
    val visit_length_7s_9s = value.getOrElse(Constants.TIME_PERIOD_7s_9s, 0)
    val visit_length_10s_30s = value.getOrElse(Constants.TIME_PERIOD_10s_30s, 0)
    val visit_length_30s_60s = value.getOrElse(Constants.TIME_PERIOD_30s_60s, 0)
    val visit_length_1m_3m = value.getOrElse(Constants.TIME_PERIOD_1m_3m, 0)
    val visit_length_3m_10m = value.getOrElse(Constants.TIME_PERIOD_3m_10m, 0)
    val visit_length_10m_30m = value.getOrElse(Constants.TIME_PERIOD_10m_30m, 0)
    val visit_length_30m = value.getOrElse(Constants.TIME_PERIOD_30m, 0)

    // 获取对应的步长的值
    val step_length_1_3 = value.getOrElse(Constants.STEP_PERIOD_1_3, 0)
    val step_length_4_6 = value.getOrElse(Constants.STEP_PERIOD_4_6, 0)
    val step_length_7_9 = value.getOrElse(Constants.STEP_PERIOD_7_9, 0)
    val step_length_10_30 = value.getOrElse(Constants.STEP_PERIOD_10_30, 0)
    val step_length_30_60 = value.getOrElse(Constants.STEP_PERIOD_30_60, 0)
    val step_length_60 = value.getOrElse(Constants.STEP_PERIOD_60, 0)

    // 计算对应的访问时长在符合条件的session中的比例
    // NumberUtils.formatDouble(小数，保留几位小数)
    val visit_length_1s_3s_ratio = NumberUtils.formatDouble(visit_length_1s_3s / session_count, 2)
    val visit_length_4s_6s_ratio = NumberUtils.formatDouble(visit_length_4s_6s / session_count, 2)
    val visit_length_7s_9s_ratio = NumberUtils.formatDouble(visit_length_7s_9s / session_count, 2)
    val visit_length_10s_30s_ratio = NumberUtils.formatDouble(visit_length_10s_30s / session_count, 2)
    val visit_length_30s_60s_ratio = NumberUtils.formatDouble(visit_length_30s_60s / session_count, 2)
    val visit_length_1m_3m_ratio = NumberUtils.formatDouble(visit_length_1m_3m / session_count, 2)
    val visit_length_3m_10m_ratio = NumberUtils.formatDouble(visit_length_3m_10m / session_count, 2)
    val visit_length_10m_30m_ratio = NumberUtils.formatDouble(visit_length_10m_30m / session_count, 2)
    val visit_length_30m_ratio = NumberUtils.formatDouble(visit_length_30m / session_count, 2)
    // 计算对应的步长在符合条件的session中的比例
    val step_length_1_3_ratio = NumberUtils.formatDouble(step_length_1_3 / session_count, 2)
    val step_length_4_6_ratio = NumberUtils.formatDouble(step_length_4_6 / session_count, 2)
    val step_length_7_9_ratio = NumberUtils.formatDouble(step_length_7_9 / session_count, 2)
    val step_length_10_30_ratio = NumberUtils.formatDouble(step_length_10_30 / session_count, 2)
    val step_length_30_60_ratio = NumberUtils.formatDouble(step_length_30_60 / session_count, 2)
    val step_length_60_ratio = NumberUtils.formatDouble(step_length_60 / session_count, 2)

    // 按照SessionAggrStat 样例对象拼接字符串，则行用+号
    val stat = SessionAggrStat(taskUUID, session_count.toInt, visit_length_1s_3s_ratio, visit_length_4s_6s_ratio, visit_length_7s_9s_ratio, visit_length_10s_30s_ratio, visit_length_30s_60s_ratio, visit_length_1m_3m_ratio, visit_length_3m_10m_ratio, visit_length_10m_30m_ratio, visit_length_30m_ratio, step_length_1_3_ratio, step_length_4_6_ratio, step_length_7_9_ratio, step_length_10_30_ratio, step_length_30_60_ratio, step_length_60_ratio)

    // 字符串需要array
    val sessionRatioRDD: RDD[SessionAggrStat] = sparkSession.sparkContext.makeRDD(Array(stat))


    import sparkSession.implicits._

    sessionRatioRDD.toDF.write
      .format("jdbc")
      .option("url", ConfigurationManager.config.getString(Constants.JDBC_URL))
      .option("user", ConfigurationManager.config.getString(Constants.JDBC_USER))
      .option("password", ConfigurationManager.config.getString(Constants.JDBC_PASSWORD))
      .option("dbtable", "session_stat_ratio_0416")
      .mode(SaveMode.Append)
      .save()

  }

  /**
   * 计算访问时长，给对应的累加器计数
   *
   * @param visitLength                 访问时长
   * @param sessionStatisticAccumulator 自定义了累加器
   */
  def calculateVisitLength(visitLength: Long, sessionStatisticAccumulator: SessionAccumulator) = {
    if (visitLength >= 1 && visitLength <= 3) {
      sessionStatisticAccumulator.add(Constants.TIME_PERIOD_1s_3s)
    } else if (visitLength >= 4 && visitLength <= 6) {
      sessionStatisticAccumulator.add(Constants.TIME_PERIOD_4s_6s)
    } else if (visitLength >= 7 && visitLength <= 9) {
      sessionStatisticAccumulator.add(Constants.TIME_PERIOD_7s_9s)
    } else if (visitLength >= 10 && visitLength <= 30) {
      sessionStatisticAccumulator.add(Constants.TIME_PERIOD_10s_30s)
    } else if (visitLength > 30 && visitLength <= 60) {
      sessionStatisticAccumulator.add(Constants.TIME_PERIOD_30s_60s)
    } else if (visitLength > 60 && visitLength <= 180) {
      sessionStatisticAccumulator.add(Constants.TIME_PERIOD_1m_3m)
    } else if (visitLength > 180 && visitLength <= 600) {
      sessionStatisticAccumulator.add(Constants.TIME_PERIOD_3m_10m)
    } else if (visitLength > 600 && visitLength <= 1800) {
      sessionStatisticAccumulator.add(Constants.TIME_PERIOD_10m_30m)
    } else if (visitLength > 1800) {
      sessionStatisticAccumulator.add(Constants.TIME_PERIOD_30m)
    }
  }

  /**
   * 计算访问步长
   *
   * @param stepLength                  访问步长
   * @param sessionStatisticAccumulator 自定义累加器
   */
  def calculateStepLength(stepLength: Long, sessionStatisticAccumulator: SessionAccumulator) = {
    if (stepLength >= 1 && stepLength <= 3) {
      sessionStatisticAccumulator.add(Constants.STEP_PERIOD_1_3)
    } else if (stepLength >= 4 && stepLength <= 6) {
      sessionStatisticAccumulator.add(Constants.STEP_PERIOD_4_6)
    } else if (stepLength >= 7 && stepLength <= 9) {
      sessionStatisticAccumulator.add(Constants.STEP_PERIOD_7_9)
    } else if (stepLength >= 10 && stepLength <= 30) {
      sessionStatisticAccumulator.add(Constants.STEP_PERIOD_10_30)
    } else if (stepLength > 30 && stepLength <= 60) {
      sessionStatisticAccumulator.add(Constants.STEP_PERIOD_30_60)
    } else if (stepLength > 60) {
      sessionStatisticAccumulator.add(Constants.STEP_PERIOD_60)
    }

  }

  // 根据条件过滤getSessionFilteredRDD(筛选条件，被筛选数据，自定义累加器)
  def getSessionFilteredRDD(taskParam: JSONObject, sessionId2FullInfoRDD: RDD[(String, String)], sessionAccumulator: SessionAccumulator) = {
    // 从json对象中获取指定参数的数值
    val startAge = ParamUtils.getParam(taskParam, Constants.PARAM_START_AGE)
    val endAge = ParamUtils.getParam(taskParam, Constants.PARAM_END_AGE)
    val professionals = ParamUtils.getParam(taskParam, Constants.PARAM_PROFESSIONALS)
    val cities = ParamUtils.getParam(taskParam, Constants.PARAM_CITIES)
    val sex = ParamUtils.getParam(taskParam, Constants.PARAM_SEX)
    val keywords = ParamUtils.getParam(taskParam, Constants.PARAM_KEYWORDS)
    val categoryIds = ParamUtils.getParam(taskParam, Constants.PARAM_CATEGORY_IDS)

    // 拼接过滤条件
    var filterInfo = (if (startAge != null) Constants.PARAM_START_AGE + "=" + startAge + "|" else "") +
      (if (endAge != null) Constants.PARAM_END_AGE + "=" + endAge + "|" else "") +
      (if (professionals != null) Constants.PARAM_PROFESSIONALS + "=" + professionals + "|" else "") +
      (if (cities != null) Constants.PARAM_CITIES + "=" + cities + "|" else "") +
      (if (sex != null) Constants.PARAM_SEX + "=" + sex + "|" else "") +
      (if (keywords != null) Constants.PARAM_KEYWORDS + "=" + keywords + "|" else "") +
      (if (categoryIds != null) Constants.PARAM_CATEGORY_IDS + "=" + categoryIds else "")

    // 去掉最后的|
    if (filterInfo.endsWith("\\|")) {
      filterInfo = filterInfo.substring(0, filterInfo.length - 1)
    }

    //  根据条件过滤数据，filter为true时返回当前的数据
    val filter2Info: RDD[(String, String)] = sessionId2FullInfoRDD.filter {
      case (sessionId, fullInfo) =>
        var success = true
        // between(数据,数据字段,参数,起始参数字段,结束参数字段)
        if (!ValidUtils.between(fullInfo, Constants.FIELD_AGE, filterInfo, Constants.PARAM_START_AGE, Constants.PARAM_END_AGE)) {
          // 如果在范围内则success还是true，如果不在的话success为false。当前的fullinfo这条数据就被滤除掉
          success = false
        } else if (!ValidUtils.in(fullInfo, Constants.FIELD_PROFESSIONAL, filterInfo, Constants.PARAM_PROFESSIONALS)) {
          success = false
        } else if (!ValidUtils.in(fullInfo, Constants.FIELD_CITY, filterInfo, Constants.PARAM_CITIES)) {
          success = false
        } else if (!ValidUtils.equal(fullInfo, Constants.FIELD_SEX, filterInfo, Constants.PARAM_SEX)) {
          success = false
        } else if (!ValidUtils.in(fullInfo, Constants.FIELD_SEARCH_KEYWORDS, filterInfo, Constants.PARAM_KEYWORDS)) {
          success = false
        } else if (!ValidUtils.in(fullInfo, Constants.FIELD_CLICK_CATEGORY_IDS, filterInfo, Constants.PARAM_CATEGORY_IDS)) {
          success = false
        }

        // 如果success为true表示sessionid对应的fullinfo满足所有的条件
        if (success) {
          // 满足条件的信息加一， session_count为k+1
          sessionAccumulator.add(Constants.SESSION_COUNT)

          // 获取当前信息的访问时长和步长
          val visitLength: Long = StringUtils.getFieldFromConcatString(fullInfo, "\\|", Constants.FIELD_VISIT_LENGTH).toLong
          val stepLength: Long = StringUtils.getFieldFromConcatString(fullInfo, "\\|", Constants.FIELD_STEP_LENGTH).toLong

          // 根据visitLength，累加计数
          calculateVisitLength(visitLength, sessionAccumulator)
          calculateStepLength(stepLength, sessionAccumulator)

        }
        success
    }
    filter2Info
  }


  def getSessionFullInfo(sparkSession: SparkSession, session2GroupActionRDD: RDD[(String, Iterable[UserVisitAction])]) = {

    // userId2AggrInfoRDD: RDD[(userId, aggrInfo)]
    val userId2AggrInfoRDD: RDD[(Long, String)] = session2GroupActionRDD.map {
      case (sessionId, iterableAction) =>
        var userId = -1L
        var startTime: Date = null
        var endTime: Date = null

        var stepLength = 0

        val searchKeywords = new StringBuffer("")
        val clickCategories = new StringBuffer("")

        for (action <- iterableAction) {

          if (userId == -1L) {
            // 如果id为-1，则把迭代器中的userid赋值
            userId = action.user_id
          }

          // 将字符串转换为日期格式
          val actionTime: Date = DateUtils.parseTime(action.action_time)

          // startTime.after(actionTime)左侧的时间是不是比右侧的时间要晚
          if (startTime == null || startTime.after(actionTime)) {
            startTime = actionTime
          }
          if (endTime == null || endTime.before(actionTime)) {
            endTime = actionTime
          }

          val searchKeyword = action.search_keyword
          // 判断 searchKeyword不为空 并且 searchKeywords里不包含searchKeyword
          if (StringUtils.isNotEmpty(searchKeyword) && !searchKeywords.toString.contains(searchKeyword)) {
            searchKeywords.append(searchKeyword + ",")
          }

          val clickCategoryId: Long = action.click_category_id
          if (clickCategoryId != -1L && !clickCategories.toString.contains(clickCategoryId)) {
            clickCategories.append(clickCategoryId + ",")
          }

          // 步长
          stepLength += 1
        }
        // 去掉字符串拼接的最后一个逗号
        val searchKw: String = StringUtils.trimComma(searchKeywords.toString)
        val clickCg: String = StringUtils.trimComma(clickCategories.toString)


        // endTime.getTime 获得的事毫秒
        val visitLength = (endTime.getTime - startTime.getTime) / 1000

        // 将聚合是值拼接成一个聚合信息
        val aggrInfo = Constants.FIELD_SESSION_ID + "=" + sessionId + "|" +
          Constants.FIELD_SEARCH_KEYWORDS + "=" + searchKw + "|" +
          Constants.FIELD_CLICK_CATEGORY_IDS + "=" + clickCg + "|" +
          Constants.FIELD_VISIT_LENGTH + "=" + visitLength + "|" +
          Constants.FIELD_STEP_LENGTH + "=" + stepLength + "|" +
          Constants.FIELD_START_TIME + "=" + DateUtils.formatTime(startTime)

        (userId, aggrInfo)

    }

    var sql = "select * from user_info"
    import sparkSession.implicits._
    // userId2InfoRDD(userid,userInfo)得到k为userid和 v为userinfo的kv结构
    val userId2InfoRDD: RDD[(Long, UserInfo)] = sparkSession.sql(sql).as[UserInfo].rdd.map(item => (item.user_id, item))

    val sessionId2FullInfoRDD: RDD[(String, String)] = userId2AggrInfoRDD.join(userId2InfoRDD).map {
      //join 后得到到数据结构就是 (userId, (aggrInfo:string, userInfo:UserInfo))
      case (userId, (aggrInfo, userInfo)) =>
        val age = userInfo.age

        val professional = userInfo.professional
        val sex = userInfo.sex
        val city = userInfo.city

        // 聚合拼接
        val fullInfo = aggrInfo + "|" +
          Constants.FIELD_AGE + "=" + age + "|" +
          Constants.FIELD_PROFESSIONAL + "=" + professional + "|" +
          Constants.FIELD_SEX + "=" + sex + "|" +
          Constants.FIELD_CITY + "=" + city

        // 从拼接的字符串中提取字段getFieldFromConcatString(字符串, 分隔符, 字段)
        val sessionId = StringUtils.getFieldFromConcatString(aggrInfo, "\\|", Constants.FIELD_SESSION_ID)

        (sessionId, fullInfo)
    }
    sessionId2FullInfoRDD
  }


  // 按照taskParam的json对象，查询条件内的数据信息并返回rdd
  def getOriActionRDD(sparkSession: SparkSession, taskParam: JSONObject) = {

    // 获取json对象的起始日期
    val startDate: String = ParamUtils.getParam(taskParam, Constants.PARAM_START_DATE)
    // 获取json对象的结束日期
    val endDate: String = ParamUtils.getParam(taskParam, Constants.PARAM_END_DATE)

    val sql = "select * from user_visit_action where date>='" + startDate + "'and date<='" + endDate + "'"
    import sparkSession.implicits._
    // row =>dateset=>rdd
    sparkSession.sql(sql).as[UserVisitAction].rdd
  }
}
